package temporay;

import temporay.message.IOWorkRequest;
import temporay.worker.WorkerService;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * program: io-share 
 * author: bsworld.xie 
 * create: 2021-04-19 16:20 
 * description: 
 */
public class SelectorThread extends AbstractSelectThread {

    private final Queue<SocketChannel> acceptedQueue;
    private final Queue<SelectionKey> updateQueue;

    private WorkerService workerPool;

    public SelectorThread(int id, WorkerService workerPool) throws IOException {
        super("NIOServerCxnFactory.SelectorThread-" + id);
        acceptedQueue = new LinkedBlockingQueue<SocketChannel>();
        updateQueue = new LinkedBlockingQueue<SelectionKey>();
        this.workerPool = workerPool;
    }

    public boolean addAcceptedConnection(SocketChannel accepted) {
        if (acceptedQueue.offer(accepted)) {
            return false;
        }
        wakeupSelector();
        return true;
    }

    public void addInterestOpsUpdateRequest(SelectionKey sk) {
        if (!updateQueue.offer(sk)) {
            return;
        }
        wakeupSelector();
    }

    public void run() {
        while (true) {
            try {
                select();
                processAcceptedConnections();
                processInterestOpsUpdateRequests();
            } catch (Exception e) {
            }
        }

    }

    //fixme io多路复用其实本质上也就是一个selector可以监听多个selectionKey，并且做相应的处理
    private void select() {
        try {
            selector.select();
            Set<SelectionKey> selected = selector.selectedKeys();
            ArrayList<SelectionKey> selectedList = new ArrayList<SelectionKey>(selected);
            Collections.shuffle(selectedList);
            Iterator<SelectionKey> selectedKeys = selectedList.iterator();
            while (selectedKeys.hasNext()) {
                SelectionKey key = selectedKeys.next();
                selected.remove(key);

                if (!key.isValid()) {
                    cleanupSelectionKey(key);
                    continue;
                }
                if (key.isReadable() || key.isWritable()) {
                    handleIO(key);
                }
            }
        } catch (IOException e) {
        }
    }

    private void handleIO(SelectionKey key) {
        IOWorkRequest workRequest = new IOWorkRequest(this, key);
        workerPool.doWork(workRequest);
    }

    //fixme socketChannel 注册到selector上,也就是一个selector上可以绑定多个selectionKey，这样处理来自客户端的读写事件
    private void processAcceptedConnections() {
        SocketChannel socketChannel;
        while ((socketChannel = acceptedQueue.poll()) != null) {
            SelectionKey key = null;
            try {
                key = socketChannel.register(selector, SelectionKey.OP_READ);
                Connection connection = new Connection(socketChannel, key, this);
                key.attach(connection);
            } catch (IOException e) {
                cleanupSelectionKey(key);
            }
        }
    }


    private void processInterestOpsUpdateRequests() {
        SelectionKey key;
        while ((key = updateQueue.poll()) != null) {
            if (!key.isValid()) {
                cleanupSelectionKey(key);
            }
            Connection cnxn = (Connection) key.attachment();
            key.interestOps(cnxn.getInterestOps());
        }
    }


}
